package util

import (
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"runtime/debug"

	"github.com/streadway/amqp"
)

/**
用法
func NotifyCloseFlow(flowList []*CloseFlowInfo) error {
	if amqp.GetClient() == nil {
		amqp.ConnectToBroker([]string{conf.Cfg.RabbitMQ.AmqpUrl})
		//amqp.ConnectToBroker([]string{"amqp://192.168.0.43:5672"})
	}
	if len(flowList) < 1 {
		return errors.New("参数错误")
	}
	err := amqp.Publish(amqp.Ctrl{
		ExchangeName: "exc_go2node_ws_" + flowList[0].CenterId,
		ExchangeType: "direct",
		RoutingKey:   "rk_traffic_flow",
		//QueueName:    "q_go_ws_111111111",
		Durable:     false,
		IsBindQueue: false,
	}, CloseNotifyData{"close_flow", flowList})
	return err
}
*/

type ConnectPool struct {
	url           []string
	connectionNum int // 连接数
	connections   map[int]*amqp.Connection
}

var Clients ConnectPool

func GetClient() *amqp.Connection {

	for _, v := range Clients.connections {
		return v
	}
	return nil
}

// url连接
func connect(amqpUrl string) *amqp.Connection {
	conn, err := amqp.Dial(amqpUrl)
	if err != nil {
		fmt.Printf(" connect amqpUrl:%s %v", amqpUrl, err)
		return nil
	}
	return conn
}

// ConnectToBroker /* amqp连接虚拟主机 */
func ConnectToBroker(url []string) {
	if len(url) == 0 {
		fmt.Printf("Cannot initialize connection to broker, connectionString not set. Have you initialized?")
		return
	}
	Clients.connections = make(map[int]*amqp.Connection)
	for i, v := range url {
		amqpClient := connect(v)
		if amqpClient == nil {
			continue
		}
		Clients.connections[i] = amqpClient
		Clients.url = append(Clients.url, v)
		Clients.connectionNum++
	}
	return
}

/*
    param说明：
	name: 交换机名称
	type: 交换机类型 direct，topic，fanout，headers
	durability: 是否需要持久化，true 为持久化
	auto delete: 当最后一个绑定到 exchange 上的队列被删除后，exchange 没有绑定的队列了，自动删除该 exchange
	internal: 当前 exchange 是否用于 rabbitMQ 内部使用，默认为 false
	arguments: 扩展参数，用于扩展 AMQP 协议自制定化使用
*/

type Ctrl struct {
	ExchangeName string // 交换机名称
	ExchangeType string // 交换机类型
	RoutingKey   string // key
	QueueName    string // 队列名
	ConsumerName string // 消费者
	IsBindQueue  bool   // 是否绑定队列
	Durable      bool   // 持久化
}

func reConnect(url []string) {
	if len(url) == 0 {
		fmt.Printf("Cannot initialize connection to broker, connectionString not set. Have you initialized?")
		return
	}
	for i, v := range url {
		amqpClient := connect(v)
		if amqpClient == nil {
			continue
		}
		Clients.connections[i] = amqpClient
	}
}

func getConnectOkClient() (*amqp.Connection, bool) {
	var conn *amqp.Connection
	var ok bool
	var noNum int
	for i := 0; i < Clients.connectionNum; i++ {
		conn, ok = Clients.connections[i]
		if !ok {
			continue
		}
		if !conn.IsClosed() {
			return conn, true
		}
		noNum++
	}
	if noNum == Clients.connectionNum {
		reConnect(Clients.url)
	}

	for i := 0; i < Clients.connectionNum; i++ {
		conn, ok = Clients.connections[i]
		if !ok {
			continue
		}
		if !conn.IsClosed() {
			return conn, true
		}
	}
	return conn, false
}

// Publish /* 推送数据 */
func Publish(pCtrl Ctrl, inData interface{}) error {
	if Clients.connectionNum == 0 {
		return nil
	}
	var conn *amqp.Connection
	conn, status := getConnectOkClient()
	if status == false {
		return errors.New("getConnectOkClient false")
	}
	err := publish(conn, pCtrl, inData)
	return err
}

/* 推送数据 */
func publish(client *amqp.Connection, pCtrl Ctrl, inData interface{}) error {
	if client == nil {
		fmt.Printf("amqp connection is nil  ")
		return nil
	}
	if reflect.ValueOf(inData).IsZero() {
		err := fmt.Errorf("playload is empty")
		return err
	}
	payload, err := json.Marshal(inData)
	if err != nil {
		fmt.Printf("%v", err)
		return err
	}
	ch, err := client.Channel() // Get a channel from the connection
	if err != nil {
		fmt.Printf("amqp channel create   err %v", err)
		return err
	}
	defer ch.Close()
	err = ch.ExchangeDeclare(pCtrl.ExchangeName, pCtrl.ExchangeType, pCtrl.Durable, false, false, false, nil)
	if err != nil {
		fmt.Printf("amqp exchange declare err %v", err)
		return err
	}
	var queue amqp.Queue
	if pCtrl.IsBindQueue == false {
		goto EXIT
	}

	//queue, err = ch.QueueDeclarePassive(pCtrl.QueueName, pCtrl.Durable, false, false, false, nil)
	//if err != nil {
	queue, err = ch.QueueDeclare(pCtrl.QueueName, pCtrl.Durable, false, false, false, nil)
	if err != nil {
		fmt.Printf("amqp queue declare err %v", err)
		return err
	}
	//}

	err = ch.QueueBind(queue.Name, pCtrl.RoutingKey, pCtrl.ExchangeName, false, nil)
	if err != nil {
		fmt.Printf("amqp queue bind err %v", err)
		return err
	}

EXIT:
	err = ch.Publish(pCtrl.ExchangeName, pCtrl.RoutingKey, false, false,
		amqp.Publishing{
			Body: payload, // Our JSON body as []byte
		})
	if err != nil {
		fmt.Printf("amqp publish err %v", err)
		return err
	}
	return nil
}

// Subscribe /* 订阅 */
func Subscribe(pCtrl Ctrl, handlerFunc func(amqp.Delivery)) error {
	client, status := getConnectOkClient()
	if status == false {
		return errors.New("getConnectOkClient false")
	}
	ch, err := client.Channel()
	if err != nil {
		fmt.Printf("amqp channel create   err %v", err)
		return err
	}
	// defer ch.Close()
	err = ch.ExchangeDeclare(pCtrl.ExchangeName, pCtrl.ExchangeType, false, false, false, false, nil)
	if err != nil {
		fmt.Printf("amqp exchange declare err %v", err)
		return err
	}
	var queue amqp.Queue
	if pCtrl.IsBindQueue == false {
		goto EXIT
	}
	queue, err = ch.QueueDeclare(pCtrl.QueueName, pCtrl.Durable, false, false, false, nil)
	if err != nil {
		fmt.Printf("amqp queue declare err %v", err)
		return err
	}

	err = ch.QueueBind(queue.Name, pCtrl.RoutingKey, pCtrl.ExchangeName, false, nil)
	if err != nil {
		fmt.Printf("amqp queue Bind err %v", err)
		return err
	}

EXIT:
	msgs, err := ch.Consume(queue.Name, pCtrl.ConsumerName, true, false, false, false, nil)
	if err != nil {
		fmt.Printf("amqp consume err %v", err)
		return err
	}
	closeChan := make(chan *amqp.Error, 1)
	notifyClose := ch.NotifyClose(closeChan) // 一旦消费者的channel有错误，产生一个amqp.Error，channel监听并捕捉到这个错误
	for {
		select {
		case e := <-notifyClose:
			return e
		case msg := <-msgs:
			handlerFunc(msg)
		}

	}
	// go consumeLoop(msgs, handlerFunc)
	return nil
}

/* 回调执行跳转 */
func consumeLoop(deliveries <-chan amqp.Delivery, handlerFunc func(d amqp.Delivery)) {
	defer func() {
		if err := recover(); err != nil {
			fmt.Printf("amqp 回调执行跳转异常：%v,consumeLoop:%v", err, string(debug.Stack()))
		}
	}()
	for d := range deliveries {
		handlerFunc(d)
	}
}
